/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.broker.processor;

import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.transaction.OperationResult;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
import org.apache.rocketmq.common.TopicFilterType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.config.BrokerRole;

/**
 * 结束事务消息处理器：负责事务消息的提交和回滚。
 */
public class EndTransactionProcessor implements NettyRequestProcessor {
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.TRANSACTION_LOGGER_NAME);

    /**
     * broker控制器
     */
    private final BrokerController brokerController;

    public EndTransactionProcessor(final BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    /**
     * 处理结束事务的请求：
     * 如果是提交/回滚，则直接处理，如果事务状态不确定，则返回null值。
     * @param ctx
     * @param request
     * @return
     * @throws RemotingCommandException
     */
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws
        RemotingCommandException {
        // 创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        // 从请求中解码出请求报文头对象
        final EndTransactionRequestHeader requestHeader =
            (EndTransactionRequestHeader) request.decodeCommandCustomHeader(EndTransactionRequestHeader.class);
        LOGGER.debug("Transaction request:{}", requestHeader);

        // 如果当前broker的角色是SLAVE，则设置响应状态码SLAVE_NOT_AVAILABLE并返回响应。
        // 因为事务消息只能在Master上执行，所以如果当前broker是SLAVE，则不能执行事务消息。
        if (BrokerRole.SLAVE == brokerController.getMessageStoreConfig().getBrokerRole()) {
            response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
            LOGGER.warn("Message store is slave mode, so end transaction is forbidden. ");
            return response;
        }

        // 从请求中获取事务消息状态
        if (requestHeader.getFromTransactionCheck()) {
            switch (requestHeader.getCommitOrRollback()) {
                // 事务消息状态不确定
                case MessageSysFlag.TRANSACTION_NOT_TYPE: {
                    LOGGER.warn("Check producer[{}] transaction state, but it's pending status."
                            + "RequestHeader: {} Remark: {}",
                        RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                        requestHeader.toString(),
                        request.getRemark());
                    // 返回null值
                    return null;
                }
                // 事务消息提交
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE: {
                    LOGGER.warn("Check producer[{}] transaction state, the producer commit the message."
                            + "RequestHeader: {} Remark: {}",
                        RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                        requestHeader.toString(),
                        request.getRemark());
                    // 跳出分支判断
                    break;
                }
                // 事务消息回滚
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: {
                    LOGGER.warn("Check producer[{}] transaction state, the producer rollback the message."
                            + "RequestHeader: {} Remark: {}",
                        RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                        requestHeader.toString(),
                        request.getRemark());
                    // 跳出分支判断
                    break;
                }
                default:
                    return null;
            }
        } else {
            switch (requestHeader.getCommitOrRollback()) {

                // 事务消息状态不确定
                case MessageSysFlag.TRANSACTION_NOT_TYPE: {
                    LOGGER.warn("The producer[{}] end transaction in sending message,  and it's pending status."
                            + "RequestHeader: {} Remark: {}",
                        RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                        requestHeader.toString(),
                        request.getRemark());
                    return null;
                }

                // 提交事务消息
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE: {
                    break;
                }

                // 回滚事务消息
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: {
                    LOGGER.warn("The producer[{}] end transaction in sending message, rollback the message."
                            + "RequestHeader: {} Remark: {}",
                        RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                        requestHeader.toString(),
                        request.getRemark());
                    break;
                }
                default:
                    return null;
            }
        }

        // 创建操作结果对象
        OperationResult result = new OperationResult();
        // 如果是事务提交
        if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getCommitOrRollback()) {
            // 通过偏移量获取HALF消息，并封装为OperationResult对象
            result = this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);
            // 如果找到了HALF消息
            if (result.getResponseCode() == ResponseCode.SUCCESS) {
                // 如果拒绝了回滚或提交，返回异常信息
                if (rejectCommitOrRollback(requestHeader, result.getPrepareMessage())) {
                    response.setCode(ResponseCode.ILLEGAL_OPERATION);
                    LOGGER.warn("Message commit fail [producer end]. currentTimeMillis - bornTime > checkImmunityTime, msgId={},commitLogOffset={}, wait check",
                            requestHeader.getMsgId(), requestHeader.getCommitLogOffset());
                    return response;
                }
                // 检查HALF消息
                RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
                // 如果状态是SUCCESS
                if (res.getCode() == ResponseCode.SUCCESS) {
                    // 将HALF消息中的被折叠起来的主题名称、消息队列ID等信息解析出来，返回内部消息对象
                    MessageExtBrokerInner msgInner = endMessageTransaction(result.getPrepareMessage());
                    // 设置系统标记
                    msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), requestHeader.getCommitOrRollback()));
                    // 设置消息队列偏移量
                    msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());
                    // 设置HALF消息的事务偏移量
                    msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());
                    // 设置存储的时间戳
                    msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());
                    // 清空消息中的"TRAN_MSG"键值对
                    MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED);

                    // 向目标主题发送提交后的消息
                    RemotingCommand sendResult = sendFinalMessage(msgInner);
                    // 删除HALF消息
                    if (sendResult.getCode() == ResponseCode.SUCCESS) {
                        this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
                    }
                    // 返回结果
                    return sendResult;
                }
                return res;
            }
        } else if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getCommitOrRollback()) {
            // 实际上仅仅是根据报文头找到HALF消息
            result = this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);
            // 如果找到了HALF消息
            if (result.getResponseCode() == ResponseCode.SUCCESS) {
                // 如果拒绝提交或回滚，则返回错误响应
                if (rejectCommitOrRollback(requestHeader, result.getPrepareMessage())) {
                    response.setCode(ResponseCode.ILLEGAL_OPERATION);
                    LOGGER.warn("Message rollback fail [producer end]. currentTimeMillis - bornTime > checkImmunityTime, msgId={},commitLogOffset={}, wait check",
                            requestHeader.getMsgId(), requestHeader.getCommitLogOffset());
                    return response;
                }
                // 检查HALF消息
                RemotingCommand res = checkPrepareMessage(result.getPrepareMessage(), requestHeader);
                if (res.getCode() == ResponseCode.SUCCESS) {
                    // 删除HALF消息
                    this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());
                }
                // 返回响应
                return res;
            }
        }
        // 设置响应码
        response.setCode(result.getResponseCode());
        // 设置响应说明
        response.setRemark(result.getResponseRemark());
        // 返回响应对象
        return response;
    }

    /**
     * If you specify a custom first check time CheckImmunityTimeInSeconds,
     * And the commit/rollback request whose validity period exceeds CheckImmunityTimeInSeconds and is not checked back will be processed and failed
     * returns ILLEGAL_OPERATION 604 error
     * @param requestHeader
     * @param messageExt
     * @return
     */
    public boolean rejectCommitOrRollback(EndTransactionRequestHeader requestHeader, MessageExt messageExt) {
        if (requestHeader.getFromTransactionCheck()) {
            return false;
        }
        long transactionTimeout = brokerController.getBrokerConfig().getTransactionTimeOut();

        String checkImmunityTimeStr = messageExt.getUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS);
        if (StringUtils.isNotEmpty(checkImmunityTimeStr)) {
            long valueOfCurrentMinusBorn = System.currentTimeMillis() - messageExt.getBornTimestamp();
            long checkImmunityTime = TransactionalMessageUtil.getImmunityTime(checkImmunityTimeStr, transactionTimeout);
            //Non-check requests that exceed the specified custom first check time fail to return
            return valueOfCurrentMinusBorn > checkImmunityTime;
        }
        return false;
    }

    @Override
    public boolean rejectRequest() {
        return false;
    }

    /**
     * 检查HALF消息
     * @param msgExt
     * @param requestHeader
     * @return
     */
    private RemotingCommand checkPrepareMessage(MessageExt msgExt, EndTransactionRequestHeader requestHeader) {
        // 创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        if (msgExt != null) {
            // 获取消息的生产组名称
            final String pgroupRead = msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
            // 如果消息的生产组名称与请求的生产组名称不一致，则返回错误
            if (!pgroupRead.equals(requestHeader.getProducerGroup())) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The producer group wrong");
                return response;
            }
            // 如果消息队列偏移量与请求的偏移量不一致，则返回错误
            if (msgExt.getQueueOffset() != requestHeader.getTranStateTableOffset()) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The transaction state table offset wrong");
                return response;
            }
            // 如果消息的物理偏移量与请求的物理偏移量不一致，则返回错误
            if (msgExt.getCommitLogOffset() != requestHeader.getCommitLogOffset()) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The commit log offset wrong");
                return response;
            }
        } else {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("Find prepared transaction message failed");
            return response;
        }
        // 设置响应码
        response.setCode(ResponseCode.SUCCESS);
        // 返回响应
        return response;
    }

    /**
     * 结束事务消息
     * @param msgExt
     * @return
     */
    private MessageExtBrokerInner endMessageTransaction(MessageExt msgExt) {
        // 新建内部消息对象
        MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
        // 设置主题：从HALF消息的用户属性中获取被折叠起来的真正目标主题名称
        msgInner.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));
        // 设置消息队列ID：从HALF消息的用户属性中获取被折叠起来的真正的目标消息队列ID
        msgInner.setQueueId(Integer.parseInt(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_QUEUE_ID)));
        // 设置消息体
        msgInner.setBody(msgExt.getBody());
        msgInner.setFlag(msgExt.getFlag());
        msgInner.setBornTimestamp(msgExt.getBornTimestamp());
        msgInner.setBornHost(msgExt.getBornHost());
        msgInner.setStoreHost(msgExt.getStoreHost());
        msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());
        msgInner.setWaitStoreMsgOK(false);

        // 设置事务ID：从HALF消息的用户属性中获取被折叠起来的唯一KEY作为事务ID
        msgInner.setTransactionId(msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
        msgInner.setSysFlag(msgExt.getSysFlag());
        TopicFilterType topicFilterType =
            (msgInner.getSysFlag() & MessageSysFlag.MULTI_TAGS_FLAG) == MessageSysFlag.MULTI_TAGS_FLAG ? TopicFilterType.MULTI_TAG
                : TopicFilterType.SINGLE_TAG;
        long tagsCodeValue = MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
        msgInner.setTagsCode(tagsCodeValue);
        MessageAccessor.setProperties(msgInner, msgExt.getProperties());
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
        MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC);
        MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID);
        // 返回内部消息对象
        return msgInner;
    }

    /**
     * 将事务消息提交到目标主题的目标消息队列
     * @param msgInner
     * @return
     */
    private RemotingCommand sendFinalMessage(MessageExtBrokerInner msgInner) {
        // 创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        // 存储消息
        final PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
        if (putMessageResult != null) {
            switch (putMessageResult.getPutMessageStatus()) {
                // Success
                case PUT_OK:
                case FLUSH_DISK_TIMEOUT:
                case FLUSH_SLAVE_TIMEOUT:
                case SLAVE_NOT_AVAILABLE:
                    response.setCode(ResponseCode.SUCCESS);
                    response.setRemark(null);
                    break;
                // Failed
                case CREATE_MAPPED_FILE_FAILED:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("Create mapped file failed.");
                    break;
                case MESSAGE_ILLEGAL:
                case PROPERTIES_SIZE_EXCEEDED:
                    response.setCode(ResponseCode.MESSAGE_ILLEGAL);
                    response.setRemark(String.format("The message is illegal, maybe msg body or properties length not matched. msg body length limit %dB, msg properties length limit 32KB.",
                        this.brokerController.getMessageStoreConfig().getMaxMessageSize()));
                    break;
                case SERVICE_NOT_AVAILABLE:
                    response.setCode(ResponseCode.SERVICE_NOT_AVAILABLE);
                    response.setRemark("Service not available now.");
                    break;
                case OS_PAGE_CACHE_BUSY:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("OS page cache busy, please try another machine");
                    break;
                case WHEEL_TIMER_MSG_ILLEGAL:
                    response.setCode(ResponseCode.MESSAGE_ILLEGAL);
                    response.setRemark(String.format("timer message illegal, the delay time should not be bigger than the max delay %dms; or if set del msg, the delay time should be bigger than the current time",
                        this.brokerController.getMessageStoreConfig().getTimerMaxDelaySec() * 1000L));
                    break;
                case WHEEL_TIMER_FLOW_CONTROL:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark(String.format("timer message is under flow control, max num limit is %d or the current value is greater than %d and less than %d, trigger random flow control",
                        this.brokerController.getMessageStoreConfig().getTimerCongestNumEachSlot() * 2L, this.brokerController.getMessageStoreConfig().getTimerCongestNumEachSlot(), this.brokerController.getMessageStoreConfig().getTimerCongestNumEachSlot() * 2L));
                    break;
                case WHEEL_TIMER_NOT_ENABLE:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark(String.format("accurate timer message is not enabled, timerWheelEnable is %s",
                        this.brokerController.getMessageStoreConfig().isTimerWheelEnable()));
                    break;
                case UNKNOWN_ERROR:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("UNKNOWN_ERROR");
                    break;
                case IN_SYNC_REPLICAS_NOT_ENOUGH:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("in-sync replicas not enough");
                    break;
                case PUT_TO_REMOTE_BROKER_FAIL:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("put to remote broker fail");
                    break;
                default:
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("UNKNOWN_ERROR DEFAULT");
                    break;
            }
            return response;
        } else {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("store putMessage return null");
        }
        return response;
    }
}
